package org.jactao.remote.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.jactao.common.LRequest;
import org.jactao.rpc.LChannel;
import org.jactao.remote.netty.codec.DecoderFactory;
import org.jactao.remote.netty.codec.EncoderFactory;

import java.util.concurrent.TimeUnit;

/**
 * Created by fanzetao on 2018/4/4.
 */
public class NettyClient extends LChannel {

    private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(
            Math.min(Runtime.getRuntime().availableProcessors() + 1, 32),
            new DefaultThreadFactory("LRPC-NettyClientWorker", true));

    private Bootstrap bootstrap;

    private Channel channel;

    public NettyClient() {
        bootstrap = new Bootstrap();
        bootstrap.group(nioEventLoopGroup)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
    }

    @Override
    public void send(LRequest request) {
        channel.writeAndFlush(request);
    }

    @Override
    public void connect() {

        bootstrap.handler(new ChannelInitializer() {

            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline()
                        .addLast("decoder", DecoderFactory.get(getSerializer()))
                        .addLast("encoder", EncoderFactory.get(getSerializer()))
                        .addLast("handler", new NettyClientHandler());

            }
        });

        ChannelFuture future = bootstrap.connect(address);

        boolean ret = future.awaitUninterruptibly(3000, TimeUnit.MILLISECONDS);

        if (ret && future.isSuccess()) {
            Channel newChannel = future.channel();
            try {
                Channel oldChannel = this.channel;
                if (oldChannel != null) {
                    oldChannel.close();
                }
            } finally {
                this.channel = newChannel;
            }
        } else if (future.cause() != null) {
            throw new RuntimeException(future.cause());
        } else {
            throw new RuntimeException("connect error ");
        }
    }

    @Override
    public boolean connected() {
        return channel != null;
    }
}
